package com.zhang.first.day04;

import com.zhang.first.utils.SensorReading;
import com.zhang.first.utils.SensorSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @title:
 * @author: zhang
 * @date: 2022/1/17 16:28
 */
public class Example1_1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .keyBy(r -> r.sensorId)
                .process(new KeyedProcessFunction<String, SensorReading, String>() {

                    //声明温度和时间状态
                    private ValueState<Integer> lastTemp;
                    private ValueState<Long> timerTs;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //去状态后端找是否存在状态变量，如果没有则初始化，如果找到了，则直接读取。所以说是单例
                        lastTemp = getRuntimeContext().getState(new ValueStateDescriptor<Integer>(
                                "lastTemp", Types.INT
                        ));

                        timerTs = getRuntimeContext().getState(new ValueStateDescriptor<Long>(
                                "timerTs", Types.LONG
                        ));
                    }

                    @Override
                    public void processElement(SensorReading value, KeyedProcessFunction<String, SensorReading, String>.Context ctx, Collector<String> out) throws Exception {
                        Integer preTemp = null;
                        if (lastTemp.value() != null) {
                            preTemp = lastTemp.value();
                        }
                        lastTemp.update(value.temperature);

                        Long ts = null;
                        if (timerTs.value() != null) {
                            ts = timerTs.value();
                        }

                        if (preTemp != null) {
                            //温度上升且没有定时器则注册定时器
                            if (value.temperature > preTemp && ts == null) {
                                Long oneSecondLater = ctx.timerService().currentProcessingTime() + 1000L;
                                ctx.timerService().registerProcessingTimeTimer(oneSecondLater);
                                timerTs.update(oneSecondLater);
                            } else if (value.temperature < preTemp && ts != null) {
                                ctx.timerService().deleteProcessingTimeTimer(ts);
                                timerTs.clear();
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, SensorReading, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect("传感器" + ctx.getCurrentKey() + "连续1s温度连续上升！！！！");
                    }
                })
                .print();


        env.execute();
    }
}
